package hit.edu.zjc.Operator;

import hit.edu.zjc.Tool.MessyCodeCheck;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashSet;

import edu.uci.ics.hyracks.api.comm.IFrameWriter;
import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.job.IOperatorDescriptorRegistry;
import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputOperatorNodePushable;


/*
 * 根据输入的ArrayList[]以及outputArity
 * 读取每条数据条目
 * 找到他们应该去的位置，对应的writer
 * 用writer写出去
 * */
public class DataSplitOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
    private static final long serialVersionUID = 1L;
    private ArrayList<HashSet<Integer>>[] SetList_array=null;
    private RecordDescriptor rDesc=null;


    public DataSplitOperatorDescriptor(IOperatorDescriptorRegistry spec, RecordDescriptor rdesc, 
    ArrayList<HashSet<Integer>>[] setlist_array,int outputArity) 
    {
        super(spec, 1, outputArity);
        rDesc=rdesc;
        for (int i = 0; i < outputArity; i++) 
          {
            recordDescriptors[i] = rDesc;
          }
        SetList_array=setlist_array;
        
    }

    @Override
    public IOperatorNodePushable createPushRuntime(final IHyracksTaskContext ctx,
            final IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions)
            throws HyracksDataException {
    	
    	final FrameTupleAccessor accessor = new FrameTupleAccessor(ctx.getFrameSize(), rDesc);
        
    	return new AbstractUnaryInputOperatorNodePushable() {
          
            private final IFrameWriter[] writers = new IFrameWriter[outputArity];
            
            private ArrayTupleBuilder tupleBuilder;

            private ByteBuffer[] outputBuffer=new ByteBuffer[outputArity];

            private FrameTupleAppender outputAppender;
            
            
            @Override
            public void close() throws HyracksDataException 
               {
	            	for(int k=0;k<outputArity;k++)
	            	{
	            		outputAppender.reset(outputBuffer[k], false);
	            		if (outputAppender.getTupleCount() > 0) {
	                        FrameUtils.flushFrame(outputBuffer[k], writers[k]);
	                    }
	            	}
	            for (IFrameWriter writer : writers) {
	                writer.close();
	            	}
               }

            @Override
            public void fail() throws HyracksDataException {
                for (IFrameWriter writer : writers) {
                    writer.fail();
                }
            }
            
            /*
             *根据输入的DataID
             *在SetList_array中找DataID所在的SetList的索引
             *存在则返回索引，不存在则返回-1
             * */
            public int SearchID(int DataID)
               {
            		for(int k=0;k<SetList_array.length;k++)
            		{
            			ArrayList<HashSet<Integer>> SetList=SetList_array[k];
            			for(int h=0;h<SetList.size();h++)
            			{
            				HashSet<Integer> set=SetList.get(h);
            				boolean isexist=set.contains(DataID);
            				if(isexist)//存在
            				{
            					return k;
            				}
            			}
            		}
            		return -1;
               }

            @Override
            public void nextFrame(ByteBuffer buffer) throws HyracksDataException 
               {
                accessor.reset(buffer);
                int tupleCount = accessor.getTupleCount();
                int fieldCount = accessor.getFieldCount();
                int fieldSlotsLength=accessor.getFieldSlotsLength();
                byte[] byte_array=accessor.getBuffer().array();
                for (int i = 0; i < tupleCount; i++) {
                	try {
                	int tupleStartOffset = accessor.getTupleStartOffset(i);//获得第i个Tuple的开始位置。
                    tupleBuilder.reset();
                    
                    	 //获取DataID
                    int fieldstartoffset=accessor.getFieldStartOffset(i, 0);
                	 int fieldendoffset=accessor.getFieldEndOffset(i, 0);
                	 int fieldlength=accessor.getFieldLength(i,0);
                	 
                	 String data_str=UTF8StringSerializerDeserializer.INSTANCE.deserialize( 
								new DataInputStream(
										new ByteArrayInputStream(byte_array,tupleStartOffset+fieldSlotsLength+fieldstartoffset,fieldlength)));
						 int Data_ID=Integer.parseInt(data_str);//记录数据条目的主键值。
						 //寻找DataID应该放置的位置
						 int partition=SearchID(Data_ID);
						 if(partition==-1)
							 continue;
						 //构建数据
                    for(int fIdx=0;fIdx<fieldCount;fIdx++)
                    	 {
                    	fieldstartoffset=accessor.getFieldStartOffset(i, fIdx);
                    	fieldendoffset=accessor.getFieldEndOffset(i, fIdx);
                    	fieldlength=accessor.getFieldLength(i, fIdx);
							data_str=UTF8StringSerializerDeserializer.INSTANCE.deserialize( new DataInputStream(new ByteArrayInputStream(byte_array,tupleStartOffset+fieldSlotsLength+fieldstartoffset,fieldlength)));
                    	tupleBuilder.addField(rDesc.getFields()[fIdx], data_str);//将最终结果写进去。
                    	 }
                              
                    outputAppender.reset(outputBuffer[partition], false);
                    if (!outputAppender.append(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray(), 0,
                            tupleBuilder.getSize())) 
                         {
                        FrameUtils.flushFrame(outputBuffer[partition], writers[partition]);
                        outputAppender.reset(outputBuffer[partition], true);
                        if (!outputAppender.append(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray(), 0,
                                tupleBuilder.getSize())) 
                              {
                            throw new HyracksDataException(
                                    "Failed to copy an record into a frame: the record size is too large.");
                              }
                         }
                    
                    } 
                catch (Exception e) 
                	{
						e.printStackTrace();//产生异常的原因是字符串编码不是UTF-8的。
					}    
                }
               }

            @Override
            public void open() throws HyracksDataException 
               {
            		
            		tupleBuilder = new ArrayTupleBuilder(rDesc.getFieldCount());
            		outputAppender = new FrameTupleAppender(ctx.getFrameSize());
            		for(int k=0;k<outputArity;k++)
            		{
	                outputBuffer[k] = ctx.allocateFrame();   
	                outputAppender.reset(outputBuffer[k], true);
                   writers[k].open();
                    }
               }

            @Override
            public void setOutputFrameWriter(int index, IFrameWriter writer, RecordDescriptor recordDesc) {
                writers[index] = writer;
            }
        };
    }
}